package tarns_result.others.classify

import org.apache.spark.rdd.RDD
import org.apache.spark.SparkContext
import common.Tools.{getHDFSDirFiles, getSC, ifDirExistsDelete, jsonFormatREC}
object NAVIFREQ_ALL_CLASSIFY_WORLDWEBJsonToREC {
  def main(args: Array[String]): Unit = {
    val sc: SparkContext =getSC("NAVIFREQ_ALL_CLASSIFY_WORLDWEBJsonToREC")
    val inputpath_sql="/tmp/InternationalData_Result/NAVIFREQ_ALL_CLASSIFY_WORLDWEB/"
    //该路径下只有一个文件，但是该文件名不规则
    //6.将json串转化成REC格式
    val inpaths: Array[String] = getHDFSDirFiles(inputpath_sql)
    val json_file: RDD[String] = sc.textFile(inpaths.mkString(","))
    val REClines:RDD[String] = jsonFormatREC(json_file)
    val outpath_result="hdfs://node01:8020/tmp/InternationalData_Result/NAVIFREQ_ALL_CLASSIFY_WORLDWEB_RESULT"
    ifDirExistsDelete(outpath_result)
    //7.将结果写入到hdfs
    REClines.repartition(1).saveAsTextFile(outpath_result)
    sc.stop()
  }
}
